package com.wan.net.core;

import com.wan.core.shutdown.ShutdownManager;
import com.wan.net.config.HostAndPort;
import com.wan.net.rpc.registry.ZookeeperRegistry;
import com.wan.util.Constants;
import com.wan.util.cglib.FastClassUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.EventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

public abstract class AbstractServer {
    protected static final Logger logger = LoggerFactory.getLogger(AbstractServer.class);
    protected static final List<AbstractServer> SERVER_LIST = new ArrayList<>(1);
    protected static final int IDLE_TIMEOUT = 180;
    static {
        ShutdownManager.dependOn(AbstractServer::shutdownAllServers, ZookeeperRegistry.class);
    }

    protected HostAndPort hostAndPort;
    // 配置服务端nio线程组，服务端接受客户端连接
    private EventLoopGroup bossGroup;
    // SocketChannel的网络读写
    protected EventLoopGroup workerGroup;
    protected ChannelFuture channelFuture;
    protected Channel channel;
    protected final AtomicBoolean startStatus;
    // 空闲超时时间（秒）
    protected final int idleTimeout;
    protected final ChannelHandler channelHandler;

    public AbstractServer(HostAndPort hostAndPort){
        this(hostAndPort,IDLE_TIMEOUT);
    }

    public AbstractServer(HostAndPort hostAndPort,int idleTimeout){
        this(hostAndPort,idleTimeout,null);
    }

    public AbstractServer(HostAndPort hostAndPort, @Nullable ChannelHandler channelHandler){
        this(hostAndPort,IDLE_TIMEOUT,channelHandler);
    }

    public AbstractServer(HostAndPort hostAndPort, int idleTimeout, @Nullable ChannelHandler channelHandler){
        this.hostAndPort = hostAndPort;
        this.startStatus = new AtomicBoolean(false);
        this.idleTimeout = idleTimeout;
        if(channelHandler == null){
            channelHandler = FastClassUtil.newInstance(getServerType().getAssignHandler());
        }
        this.channelHandler = channelHandler;
    }

    protected abstract void channelChannelInitializer(ChannelPipeline pipeline);

    public synchronized void start() {
        if(!startStatus.compareAndSet(false,true)){
            return;
        }
        int cpuNum = Runtime.getRuntime().availableProcessors();
        bossGroup = Epoll.isAvailable()
                ? new EpollEventLoopGroup(Math.max(1, cpuNum / 4), new DefaultThreadFactory("netty-boss", true))
                : new NioEventLoopGroup(Math.max(1, cpuNum / 4), new DefaultThreadFactory("netty-boss", true));

        workerGroup = Epoll.isAvailable()
                ? new EpollEventLoopGroup(cpuNum * 2, new DefaultThreadFactory("netty-worker", true))
                : new NioEventLoopGroup(cpuNum * 2, new DefaultThreadFactory("netty-worker", true));

        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .channel(Epoll.isAvailable() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .option(ChannelOption.SO_REUSEADDR, true)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(16 * Constants.BYTE_OF_KB, 16 * Constants.BYTE_OF_MB))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new IdleStateHandler(0,0,idleTimeout));
                        channelChannelInitializer(pipeline);
                        //消息处理器
                        pipeline.addLast(channelHandler);
                    }
                });
        // 绑定端口，同步等待成功
        // channelFuture = bootstrap.bind(hostAddress, port).sync();
        // 等待服务端监听端口关闭
        // channelFuture.channel().closeFuture().sync();


        // 异步
        channelFuture = bootstrap.bind(hostAndPort.getHost(), hostAndPort.getPort());
        channelFuture.syncUninterruptibly();
        channel = channelFuture.channel();
        SERVER_LIST.add(this);
        logger.info("{} server started at [{}]",getServerType().name().toLowerCase() ,hostAndPort);
    }


    public synchronized void shutdown() {
        if(!startStatus.get()){
            return;
        }
        shutdownEventLoopGracefully(bossGroup);

        shutdownEventLoopGracefully(workerGroup);

        if (channelFuture != null) {
            try {
                channelFuture.channel().close().syncUninterruptibly();
            } catch (Exception e) {
                logger.warn(e.getMessage(), e);
            }
        }

        if (channel != null) {
            try {
                channel.close();
            } catch (Exception e) {
                logger.warn(e.getMessage(), e);
            }
        }
    }

    public static void shutdownEventLoopGracefully(EventExecutorGroup executor) {
        if (executor == null) {
            return;
        }
        try {
            if (executor.isShutdown() || executor.isTerminated()) {
                executor.shutdownGracefully();
            }
        } catch (Exception e) {
            logger.error("EventLoop Thread pool [{}] is failed to shutdown! ", executor, e);
            return;
        }
        logger.info("EventLoop Thread pool [{}] shuts down gracefully.", executor);
    }

    public abstract ServerType getServerType();

    public synchronized static void shutdownAllServers() {
        SERVER_LIST.forEach(AbstractServer::shutdown);
    }


}
